#include "Peer.h"
#include "Jobs.h"

Peer::Peer(string ip, int port, int capacity){
    // use /tmp/ for default
    _localDirectory = "downloads";
    // verify if local directory exists 
    struct stat st;
    if(stat(_localDirectory.c_str(),&st) == 0){
        DBG_PRINTF((" directory is present\n"));
    } else {   
        mkdir(_localDirectory.c_str(), 0750);
    }

    localInfo = new peerInfo();
    localInfo->IP = ip;
    localInfo->Port = port;
    localInfo->State = peerInfo::unknown;
    localInfo->capacity = capacity;
    localInfo->totalHostedFiles = 0;

    _sharing = false;
    _rejoin = false;

    _peerList = "peerList.txt";

    sem_init (&share_lock, 0, 0);

    pthread_mutex_init (&transit_mutex, NULL);
    pthread_mutex_init (&job_queue_mutex, NULL);
    pthread_mutex_init (&status_mutex, NULL);
    pthread_mutex_init (&sharing_mutex, NULL);
    pthread_mutex_init (&file_writing_mutex, NULL);
    pthread_mutex_init (&host_machine_list_mutex, NULL);
}

// delete any remaining job, destroy all threads, delete transit vector
Peer::~Peer(){
    // send an EXIT job to the listener and end it
    Job *job = new Job();
    job->type = Job::EXITLISTENER;
    job->args = NULL;
    pushJob(job);
    
    pthread_join(_listenerContext, NULL);
    DBG_PRINTF(("listener ended\n"));
    pthread_cancel(_listenerContext);

    // send EXIT job to all workers to end them
    for(int i = 0; i < maxWorkers; i++){
        Job *endworker = new Job();
        endworker->type = Job::EXITWORKER;
        endworker->args = NULL;
        pushJob(endworker); 
    }

    for (unsigned int ii = 0; ii < _workerContext.size(); ii++) {
        pthread_join(*_workerContext[ii], NULL);
    }

    // delete all thread pointers
    while (!_workerContext.empty()) {
        delete _workerContext.back();
        _workerContext.pop_back();
    }


    // destroy the semaphore
    sem_destroy (&job_queue_count);

    sem_destroy (&share_lock);
    pthread_mutex_destroy(&transit_mutex);
    pthread_mutex_destroy(&job_queue_mutex);
    pthread_mutex_destroy(&status_mutex);
    pthread_mutex_destroy(&sharing_mutex);
    pthread_mutex_destroy(&file_writing_mutex);
    pthread_mutex_destroy(&host_machine_list_mutex);

    for (int ii = 0; ii < _numPeers; ii++) {
        delete _peers[ii];
    }

    pthread_mutex_destroy (&numConnected_mutex);
    pthread_mutex_destroy (&index_mutex);
    pthread_mutex_destroy (&peersArray_mutex);
}

void Peer::display(){
    pthread_mutex_lock(&status_mutex);
    _status.display();
    pthread_mutex_unlock(&status_mutex);
    if(localInfo->State == peerInfo::connected){
        displayNumPeers();
    }
}

int Peer::process_message(int client_sock, char *buf, int bufLen) {
    // parse buf
    char *field, *brkt;
    int msgType;
    int status = SAFE;

    if (bufLen < 3 || buf[0] != '^' || buf[bufLen-2] != '$') return errInvalidMessage; // return errInvalidMessage error

    field = strtok_r(buf, "^%$", &brkt); // get message type
    msgType = atoi(field);
    DBG_PRINTF(("msg type: %i\n", msgType));

    switch(msgType){
        case JOIN: {
            close(client_sock);

            string ip = "";
            int port = 0;
            int capacity = 0;
            // continue parsing
            field = strtok_r(NULL, "^%$", &brkt); // get IP addr
            ip += field;

            field = strtok_r(NULL, "^%$", &brkt); // get Port #
            port = atoi(field);

            field = strtok_r(NULL, "^%$", &brkt); // get capacity
            capacity = atoi(field);

            DBG_PRINTF(("received JOIN: %s:%i with capacity = %i\n", ip.c_str(), port, capacity));

            peerJoin(ip, port, capacity);

            break;
        }
        case LEAVE:{
            close(client_sock);

            string ip = "";
            int port = 0;
            string fileName = "";

            // continue parsing
            field = strtok_r(NULL, "^%$", &brkt); // get IP addr
            ip += field;

            field = strtok_r(NULL, "^%$", &brkt); // get Port #
            port = atoi(field);

            DBG_PRINTF(("received LEAVE: %s:%i\n", ip.c_str(), port));

            // set peer status to disconnected
            peerLeave(ip, port);

            break;
        }
        case QUERYSTATUS: {
            DBG_PRINTF(("received QUERYSTATUS\n"));

            // create REPLYSTATUS job

            QueryStatusArgs *jobArgs = new QueryStatusArgs();        
            jobArgs->clntSocket = client_sock;

            Job *job = new Job();
            job->type = Job::REPLYSTATUS;
            job->args = (void *)jobArgs;

            // acquire the job queue mutex; enqueue; release mutex
            pushJob(job);
            break;
        }
        case NEWFILE: {
            close(client_sock);

            char *name;
            int num_versions;
            string ip = "";
            int port = 0;

            // continue parsing
            field = strtok_r(NULL, "^%$", &brkt); // get IP addr
            ip += field;

            field = strtok_r(NULL, "^%$", &brkt); // get Port #
            port = atoi(field);

            name = strtok_r(NULL, "^%$", &brkt); // get file name

            field = strtok_r(NULL, "^%$", &brkt); // get # chunks in file
            num_versions = atoi(field);

            DBG_PRINTF(("received NEWFILE notification: %s [%i versions]\n", name, num_versions));

            // [*] add new file to VFS (current version is nil - haven't gottten it yet)
            
            // [*] update file hosting information
            peerNewFile(ip, port);

            break;
        }
        case SYSREQUEST: {
            DBG_PRINTF(("received SYSREQUEST\n"));

            // create REPLYSYSINFO job

            ReplySysInfoArgs *jobArgs = new ReplySysInfoArgs();        
            jobArgs->clntSocket = client_sock;

            Job *job = new Job();
            job->type = Job::REPLYSYSINFO;
            job->args = (void *)jobArgs;

            // acquire the job queue mutex; enqueue; release mutex
            pushJob(job);
            break;
        }
        case SAVESUCCESS: {
            close(client_sock);

            DBG_PRINTF(("received SAVESUCCESS\n"));

            char *name;
            int tag, revision;
            string ip = "";
            int port = 0;

            // continue parsing
            field = strtok_r(NULL, "^%$", &brkt); // get IP addr
            ip += field;

            field = strtok_r(NULL, "^%$", &brkt); // get Port #
            port = atoi(field);

            name = strtok_r(NULL, "^%$", &brkt); // get file name

            field = strtok_r(NULL, "^%$", &brkt); // get tag number
            tag = atoi(field);

            field = strtok_r(NULL, "^%$", &brkt); // get revision number
            revision = atoi(field);

            // [*] update VFS with current file status = new revision number

            break;
        }
        case TAGSUCCESS: {
            close(client_sock);

            DBG_PRINTF(("received TAGSUCCESS\n"));

            char *name;
            int tag, revision;
            string ip = "";
            int port = 0;

            // continue parsing
            field = strtok_r(NULL, "^%$", &brkt); // get IP addr
            ip += field;

            field = strtok_r(NULL, "^%$", &brkt); // get Port #
            port = atoi(field);

            name = strtok_r(NULL, "^%$", &brkt); // get file name

            field = strtok_r(NULL, "^%$", &brkt); // get tag number
            tag = atoi(field);

            field = strtok_r(NULL, "^%$", &brkt); // get revision number
            revision = atoi(field);

            // [*] update VFS with new file tag number, at revision number (likely 0)

            break;
        }
        case REQUESTFILE: {
            DBG_PRINTF(("received REQUESTFILE\n"));

            // create TRANSFERFILE job

            String name;
            int tag;
            string ip = "";
            int port = 0;

            // continue parsing
            field = strtok_r(NULL, "^%$", &brkt); // get IP addr
            ip += field;

            field = strtok_r(NULL, "^%$", &brkt); // get Port #
            port = atoi(field);

            field = strtok_r(NULL, "^%$", &brkt); // get file name
            name += field;

            field = strtok_r(NULL, "^%$", &brkt); // get tag number
            tag = atoi(field);

            TransferFileArgs *jobArgs = new TransferFileArgs();        
            jobArgs->clntSocket = client_sock;
            jobArgs->fileName = name;
            jobArgs->tagNumber = tag;

            Job *job = new Job();
            job->type = Job::TRANSFERFILE;
            job->args = (void *)jobArgs;

            // acquire the job queue mutex; enqueue; release mutex
            pushJob(job);

            break;
        }
        case REQUESTUPDATE: {
            DBG_PRINTF(("received REQUESTUPDATE\n"));

            // create TRANSFERUPDATE job

            String name;
            int tag;
            string ip = "";
            int port = 0;

            // continue parsing
            field = strtok_r(NULL, "^%$", &brkt); // get IP addr
            ip += field;

            field = strtok_r(NULL, "^%$", &brkt); // get Port #
            port = atoi(field);

            field = strtok_r(NULL, "^%$", &brkt); // get file name
            name += field;

            field = strtok_r(NULL, "^%$", &brkt); // get tag number
            tag = atoi(field);

            TransferUpdateArgs *jobArgs = new TransferUpdateArgs();        
            jobArgs->clntSocket = client_sock;
            jobArgs->fileName = name;
            jobArgs->tagNumber = tag;

            Job *job = new Job();
            job->type = Job::TRANSFERUPDATE;
            job->args = (void *)jobArgs;

            // acquire the job queue mutex; enqueue; release mutex
            pushJob(job);

            break;
        }
            case SAVEFILE: {
                // this is done by the VFS
                // need to send: type of msg, IP, Port, filename, current version of file (server needs to verify current version is up to date)
                break;
            }
            case TAGFILE: {
                // this is done by the VFS
                // need to send: type of msg, IP, Port, filename, current version of file (server needs to verify current version is up to date)
                break;
            }
            case NEWFILE: {
                NewFileArgs *params = (NewFileArgs *)msgParams;
                if(params != NULL){
                    length += sprintf(mail, "^%i%%%s%%%i%%%s%%%i", type, localInfo->IP.c_str(), localInfo->Port, params->fileName.c_str(), params->num_versions);

                    for (int ii = 0; ii < params->num_versions; ii++) {
                        length += sprintf(mail + length, "%%%i", params->tag[ii]);
                        length += sprintf(mail + length, "%%%i", params->revision[ii]);
                    }
                    length += sprintf(mail + length, "$");
                }
                break;       
            }

        case NEWVERSION: {
            close(client_sock);

            DBG_PRINTF(("received NEWVERSION\n"));

            char *name;
            int tag, revision;
            string ip = "";
            int port = 0;

            // continue parsing
            field = strtok_r(NULL, "^%$", &brkt); // get IP addr
            ip += field;

            field = strtok_r(NULL, "^%$", &brkt); // get Port #
            port = atoi(field);

            name = strtok_r(NULL, "^%$", &brkt); // get file name

            field = strtok_r(NULL, "^%$", &brkt); // get tag number
            tag = atoi(field);

            field = strtok_r(NULL, "^%$", &brkt); // get revision number
            revision = atoi(field);

            // [*] update VFS with new server tag with revision number probably (0) -- note current revision still undef


            break;
        } 
        case NEWMODIFIED: {
            close(client_sock);

            DBG_PRINTF(("received NEWMODIFIED\n"));

            char *name;
            int tag, revision;
            string ip = "";
            int port = 0;

            // continue parsing
            field = strtok_r(NULL, "^%$", &brkt); // get IP addr
            ip += field;

            field = strtok_r(NULL, "^%$", &brkt); // get Port #
            port = atoi(field);

            name = strtok_r(NULL, "^%$", &brkt); // get file name

            field = strtok_r(NULL, "^%$", &brkt); // get tag number
            tag = atoi(field);

            field = strtok_r(NULL, "^%$", &brkt); // get revision number
            revision = atoi(field);

            // [*] update VFS with new server revision number


            break;
        } 
        case REPLYSTATUS: {
            close(client_sock);

            DBG_PRINTF(("received REPLYSTATUS\n"));

            int hostedFiles, capacity;
            string ip = "";
            int port = 0;

            // continue parsing
            field = strtok_r(NULL, "^%$", &brkt); // get IP addr
            ip += field;

            field = strtok_r(NULL, "^%$", &brkt); // get Port #
            port = atoi(field);

            field = strtok_r(NULL, "^%$", &brkt); // get storage capacity
            hostedFiles = atoi(field);

            field = strtok_r(NULL, "^%$", &brkt); // get revision number
            capacity = atoi(field);

            updatePeerStatus(ip, port, hostedFiles, capacity);
        }
        case EXITLISTENER:
            return exitListener;
        default:
            close(client_sock);
            cout<<"invalid jobtype: "<< msgType <<endl;
            break;
    }

    return status;

}

// [*] instead of while loop for bind, just restart entire connection process
int Peer::listenerImplementation() {
    DBG_PRINTF(("starting listener\n"));

    int status = SAFE;

    // opens server socket
    int         server_sock,          // Original socket in server
                    client_sock;           // New socket from connect
    socklen_t       clnt_len;           // Length of client address
    struct sockaddr_in                  // Internet addr client & server
                    clnt_adr, serv_adr;
    int             len;             // Misc counters, etc.

    // SOCKET creation 
    if ((server_sock = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
        perror("generate error");
    }

    memset( &serv_adr, 0, sizeof(serv_adr) );      // Clear structure
    serv_adr.sin_family      = AF_INET;            // Set address type
    serv_adr.sin_addr.s_addr = htonl(INADDR_ANY);  // Any interface
    serv_adr.sin_port        = htons( localInfo->Port );        // Use our fake port

    // BIND
    while (bind( server_sock, (struct sockaddr *) &serv_adr,
                sizeof(serv_adr)) < 0){
        perror("bind error");
        DBG_PRINTF(("listener: trying to bind again...\n"));
    }

    // LISTEN
    if (listen(server_sock, 5) < 0 ) {
        perror("listen error");
        close (server_sock);
        return 4;
    }                   
    
        // accepts incoming client msgs
    do {
        DBG_PRINTF(("listening indefinitely\n"));

        clnt_len = sizeof(clnt_adr); 
        if ((client_sock = accept( server_sock, (struct sockaddr *) &clnt_adr,
                        &clnt_len)) < 0) {
            perror("accept error");
            close(server_sock);
            return 5;
        }

        // read from client socket
        char buf[bufSize];
        if ( (len=read(client_sock, buf, bufSize)) > 0 ){
            DBG_PRINTF(("received msg %s\n", buf));
            status = process_message (client_sock, buf, len);
            if ( status == exitListener ) {
                break;
            } else if ( status > 0 ) {
                printf("process_message() returned WARNING 0x%x\n", status);
            }
        }
    } while( 1 );

    close(server_sock);

    DBG_PRINTF(("terminating listener\n"));

    return status;
}

int Peer::getChunk(char *fileName, int chunkNum, char *result, int &size) {
    int status = SAFE;
    string absolute = _localDirectory+"/"+fileName;
    string partial = _localDirectory+"/"+fileName+".part"+convertInt(chunkNum);
    pthread_mutex_lock(&file_writing_mutex);

    ifstream ifile(absolute.c_str());
    ifstream ichunk(partial.c_str());
    if(ichunk){
        // file does not exist in complete form, partial chunk exists
        ichunk.read(result, chunkSize);
        size = chunkSize;
    } else if(ifile){
        // get length of file:
        ifile.seekg (0, ios::end);
        long length = ifile.tellg();
        ifile.seekg (0, ios::beg);

        // check if chunkNum is valid
        if(length/chunkSize+1 < chunkNum){
            status = errChunkNotFound;
        } else {
 
            // take out all the chunks before desired chunk
            for(int i = 1; i < chunkNum; i++){
                // read data as a block:
                ifile.read (result,chunkSize);
                length -= chunkSize;
            }

            // now read the desired chunk
            if(length < chunkSize){
                ifile.read(result, length);
                size = length;
            }else{
                ifile.read(result, chunkSize);
                size = chunkSize;
            }

            ifile.close();
        }
    } else {
        // file nor chunk exists
        status = errNoFileFound;
    }

    pthread_mutex_unlock(&file_writing_mutex);
    return status;
}

bool Peer::isChunkInProgress(string name, int chunk) {
    bool status = false;

    pthread_mutex_lock (&transit_mutex);
    for (int ii = 0; ii < (int)transit.size(); ii++) {
        if (transit[ii]->fileName == name && transit[ii]->chunk == chunk) {
            status = true;
            break;
        }
    }
    pthread_mutex_unlock (&transit_mutex);

    return status;
}

// !!! this should not be called unless isChunkInProgress returns false
int Peer::newChunkInProgress(string name, int chunk) {
    ChunkRequestArgs *args = new ChunkRequestArgs();
    args->fileName = name;
    args->chunk = chunk;

    pthread_mutex_lock (&transit_mutex);
    transit.push_back(args);
    pthread_mutex_unlock (&transit_mutex);

    return SAFE;
}

int Peer::chunkTransferCompleted(string name, int chunk) {

    pthread_mutex_lock (&transit_mutex);
    for (int ii = 0; ii < (int)transit.size(); ii++) {
        if (transit[ii]->fileName == name && transit[ii]->chunk == chunk) {
            delete transit[ii];           
            transit.erase(transit.begin() + ii);
            break;
        }
    }
    pthread_mutex_unlock (&transit_mutex);

    return SAFE;
}

int Peer::process_job(Job *job) {
    int status = SAFE;

    // depending on job type, connect to peer mailbox or just reply using passed in client socket

    // parse job args depending on job type
    switch(job->type){
        case Job::BROADCAST_JOIN:{
            char msg[bufSize];
            int msgLen;

            DBG_PRINTF(("processing BROADCAST_JOIN job...\n"));

            // create join message
            status = createMessage(Peer::JOIN, job->args, msg, msgLen);

            // broadcast join message to all peers
            broadcast(msg, msgLen);

            break;
        }
        case Job::BROADCAST_LEAVE:{
            char msg[bufSize];
            int msgLen;

            DBG_PRINTF(("processing BROADCAST_LEAVE job...\n"));

            // create join message
            status = createMessage(Peer::LEAVE, job->args, msg, msgLen);

            // broadcast join message to all peers
            broadcast(msg, msgLen);

            break;
        }
        case Job::QUERYCHUNK: {
            char msg[bufSize];
            int msgLen;

            DBG_PRINTF(("processing QUERYCHUNK job...\n"));

            // create join message
            status = createMessage(Peer::QUERYCHUNK, job->args, msg, msgLen);

            queryPeers(msg, msgLen, this);

            break;
        }
        case Job::WANTCHUNK: {
            char msg[bufSize];
            char buf[chunkSize];            
            int msgLen;
            int responseLen;
            int clnt_socket;
            string file;
            int chunk;

            DBG_PRINTF(("processing WANTCHUNK job...\n"));

            // create wantchunk message
            status = createMessage(Peer::WANTCHUNK, job->args, msg, msgLen);

            clnt_socket = ((ReplyChunkArgs *)(job->args))->clntSocket;
            file = ((ReplyChunkArgs *)(job->args))->fileName;
            chunk = ((ReplyChunkArgs *)(job->args))->chunk;

            write( clnt_socket, msg, msgLen );

            // wait for message containing chunk
            if (( responseLen = read(clnt_socket, buf, chunkSize)) > 0 ) {

                DBG_PRINTF(("RECEIVED CHUNK CONTENT: FILE %s CHUNK %i\n", file.c_str(), chunk));


                // create broadcast_chunk received
                ReceivedChunkArgs *jobArgs = new ReceivedChunkArgs();
                jobArgs->fileName = file;
                jobArgs->chunk = chunk;

                Job *newJob = new Job();
                newJob->type = Job::BROADCAST_RECEIVEDCHUNK;
                newJob->args = (void *)jobArgs;

                // acquire the job queue mutex; enqueue; release mutex
                pushJob(newJob);

                DBG_PRINTF(("broadcast job created: FILE %s CHUNK %i\n", file.c_str(), chunk));

                DBG_PRINTF(("writing file: %s for chunk %i\n", file.c_str(), chunk));

                // write chunk to local directory
                writeChunk(file, chunk, buf, responseLen);

                DBG_PRINTF(("now mutate own status: FILE %s CHUNK %i\n", file.c_str(), chunk));
                addChunk(file, chunk);

                // transfer completed; delete entry from transit vector
                chunkTransferCompleted( file, chunk );

                // wake up share
                sem_post (&share_lock); 
            }

            break;
        }
        case Job::BROADCAST_RECEIVEDCHUNK: {
            char msg[bufSize];           
            int msgLen;

            DBG_PRINTF(("processing BROADCAST_RECEIVEDCHUNK job...\n"));

            // create join message
            status = createMessage(Peer::RECEIVEDCHUNK, job->args, msg, msgLen);

            // broadcast join message to all peers
            broadcast(msg, msgLen);

            break;
        }
        case Job::DONTWANTCHUNK: {
            char msg[bufSize];
            int msgLen;

            DBG_PRINTF(("processing DONTWANTCHUNK job...\n"));

            // create join message
            status = createMessage(Peer::DONTWANTCHUNK, job->args, msg, msgLen);

            write( ((ReplyChunkArgs *)(job->args))->clntSocket, msg, msgLen );
            break;
        }
        case Job::QUERYSTATUS:
            // do not use this one, we use Peers::initialize instead 
            break;
        case Job::REPLYSTATUS: {

            int msgLen = 0;            
            char msg[bufSize];

            DBG_PRINTF(("processing REPLYSTATUS job...\n"));

            createMessage(Peer::REPLYSTATUS, job->args, msg, msgLen);

            write( ((QueryStatusArgs *)(job->args))->clntSocket, msg, msgLen );

            break;
        }
        case Job::BROADCAST_NEWFILE: {
            char msg[bufSize];
            int msgLen = 0;

            DBG_PRINTF(("processing BROADCAST_NEWFILE job...\n"));

            // create NEWFILE message
            status = createMessage(Peer::NEWFILE, job->args, msg, msgLen);

            // broadcast join message to all peers
            broadcast(msg, msgLen);

            break;
        }
        case Job::EXITLISTENER: {
            char msg[bufSize];
            int msgLen = 0;

            DBG_PRINTF(("processing EXITLISTENER job...\n"));

            // create NEWFILE message
            status = createMessage(Peer::EXITLISTENER, job->args, msg, msgLen);

            // send to self function
            sendMessage(localInfo->IP, localInfo->Port, msg, msgLen);

            break;
        }
        case Job::EXITWORKER: {
            status = exitWorker;
            break;
        }
        default:
            status = errInvalidJob;  // error: unspecified job type
            break;
    }

    delete job;
    return status;
}

int Peer::workerImplementation() {
    Job *job = NULL;

    while (true) {
        sem_wait (&job_queue_count);
        pthread_mutex_lock (&job_queue_mutex);
        if( !jobs.empty() ){
            job = jobs.front();
            DBG_PRINTF(("retrieving job...\n"));
            jobs.pop(); 
        }
        pthread_mutex_unlock (&job_queue_mutex);

        int status = process_job(job);
        if(status == exitWorker)break;
    }
    DBG_PRINTF(("WORKER EXITS\n"));

    return 0;
}

int Peer::sendMessage (string IP, int port, char * message, int messageLen){
    int status = errUnknownWarning;

    int             orig_sock;           // Original socket in client
    struct sockaddr_in
        serv_adr;            // Internet addr of server
    struct hostent  *host;               // The host (server) info 

    host = gethostbyname( IP.c_str() );       // Obtain host (server) info
            if (host == (struct hostent *) NULL ) {
                DBG_PRINTF(("host name not found for %s:%i\n", IP.c_str(), port));
                perror("gethostbyname ");
            } else {
                if ((orig_sock = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
                    perror("generate error");
                    return 3;
                }

                memset(&serv_adr, 0, sizeof( serv_adr));       // Clear structure
                serv_adr.sin_family = AF_INET;                 // Set address type
                memcpy(&serv_adr.sin_addr, host->h_addr, host->h_length);
                serv_adr.sin_port   = htons( port );

                // try to connect 
                if (connect( orig_sock,(struct sockaddr *)&serv_adr,
                    sizeof(serv_adr)) < 0) {
                    DBG_PRINTF(("cannot connect to peer %s:%i\n", IP.c_str(), port));
                    perror("connect error");
                } else {
                    // write message
                    write(orig_sock, message, messageLen);
                    close(orig_sock);
                }
            } 

    return status;
}

// compiler threw error when I used Peer::compareFile
bool compareFile(File * i, File * j){
    return i->minimumChunkReplicationLevel() < j->minimumChunkReplicationLevel();
}

// [*] create getMinRepLevel in status? 
int Peer::getNextChunkToShare(string &fileName, int &chunk) {
    DBG_PRINTF(("calling getNextChunkToShare\n"));

    int status = errNoChunkToShare;

    pthread_mutex_lock (&status_mutex);

    // the idea is to populate a list of chunks that exists locally, sorted by increasing chunk replication level 
    // first call the populate function from status
    _status.processListofChunks();
    Chunk share;
    int size = _status.sizeOfListOfChunks();
    DBG_PRINTF(("size of chunk list: %i\n",size));
    for(int i = 0; i < size; i++){
        status = _status.getFromListofChunks(share);
        if(status == SAFE){
            DBG_PRINTF(("file %s chunk %i has repLevel %i\n", share.fileName.c_str(), share.chunk, share.repLevel));
            if (share.repLevel < numConnected() + 1){
                fileName = share.fileName;
                chunk = share.chunk;
                break;
            }else{
                status = errNoChunkToShare;
            }
        }
        DBG_PRINTF(("size: %i, i:%i\n", size, i));
    }

    pthread_mutex_unlock (&status_mutex);

    DBG_PRINTF(("getNextChunkToShare status: %i\n", status));

    return status;
}

// [*] should not continue trying to share the same file if it has been rejected.
int Peer::shareImplementation() {
    DBG_PRINTF(("starting share\n"));

    string fileName = "";
    int chunk = 0;

    while ( 1 ) {
        DBG_PRINTF(("share waiting\n"));
        sem_wait (&share_lock);
        DBG_PRINTF(("share woke up\n"));
        // after woken up, check if something is in the process of being shared.
        // if so, sleep to wait for current sharing to be finished
        pthread_mutex_lock (&sharing_mutex);
        bool sharing = _sharing;
        pthread_mutex_unlock (&sharing_mutex);
        if (!sharing){
            if ( getNextChunkToShare(fileName, chunk) == SAFE ) {
                // create chunk broadcast job
                ChunkRequestArgs *jobArgs = new ChunkRequestArgs();        
                jobArgs->fileName = fileName;
                jobArgs->chunk = chunk; 

                Job *job = new Job();
                job->type = Job::QUERYCHUNK;
                job->args = (void *)jobArgs;

                // set _sharing to true. some chunk is going through sharing process
                setShare(true);

                // acquire the job queue mutex; enqueue; release mutex
                pushJob(job);
            }
        }
    }

    DBG_PRINTF(("terminating share\n"));

    return 0;
}

void * Peer::listener(void *This) {
    ((Peer *)This)->listenerImplementation();
    return NULL;
     
}

void * Peer::worker(void *This) {
    ((Peer *)This)->workerImplementation();

    return NULL;
}

void * Peer::share(void *This) {
    ((Peer *)This)->shareImplementation();

    return NULL;
}

int Peer::createMessage(Message type, void *msgParams, char *mail, int &length){
    // if type is GOTFILE, NEEDFILE, OR IHAVEFILE, then there is valid filenumber and chunknumber
    length = 0;

    if(mail != NULL){
        switch(type){
            case JOIN:
                length += sprintf(mail, "^%i%%%s%%%i%%%i$", type, localInfo->IP.c_str(), localInfo->Port, localInfo->capacity);
                break;
            case LEAVE:
            case SYSREQUEST:            
                length += sprintf(mail, "^%i%%%s%%%i$", type, localInfo->IP.c_str(), localInfo->Port);
                break;
            case SAVESUCCESS: 
            case TAGSUCCESS: {
                FileACKArgs *params = (FileACKArgs *)msgParams;
                if (params != NULL) {
                    length += sprintf(mail, "^%i%%%s%%%i%%%s%%%i%%%i$", type, localInfo->IP.c_str(), 
                                                                        localInfo->Port, params->fileName.c_str(), params->tag, params->revision);
                }
                break;
            }
            case REQUESTFILE: {
                // this is done by the VFS
                // need to send: type of msg, IP, Port, filename, tag
                break;
            }
            case REQUESTUPDATE: {
                // this is done by the VFS
                // need to send: type of msg, IP, Port, filename, tag, current version of file
                break;
            }
            case SAVEFILE: {
                // this is done by the VFS
                // need to send: type of msg, IP, Port, filename, current version of file (server needs to verify current version is up to date)
                break;
            }
            case TAGFILE: {
                // this is done by the VFS
                // need to send: type of msg, IP, Port, filename, current version of file (server needs to verify current version is up to date)
                break;
            }
            case NEWFILE: {
                NewFileArgs *params = (NewFileArgs *)msgParams;
                if(params != NULL){
                    length += sprintf(mail, "^%i%%%s%%%i%%%s%%%i", type, localInfo->IP.c_str(), localInfo->Port, params->fileName.c_str(), params->num_versions);

                    for (int ii = 0; ii < params->num_versions; ii++) {
                        length += sprintf(mail + length, "%%%i", params->tag[ii]);
                        length += sprintf(mail + length, "%%%i", params->revision[ii]);
                    }
                    length += sprintf(mail + length, "$");
                }
                break;       
            }
            case NEWVERSION:
            case NEWMODIFIED: {
                NewVersionArgs *params = (NewVersionArgs *)msgParams;
                if(params != NULL){
                    length = sprintf(mail, "^%i%%%s%%%i%%%s%%%i%%%i$", type, localInfo->IP.c_str(), localInfo->Port, 
                                                                       params->fileName.c_str(), params->tag, params->revision);
                }
                break;       
            }
          
            case REPLYSTATUS: {
                // get file system -> # hosted files

                // need to send: type of msg, IP, Port, # hosted files, storage capacity 
                break;
            }
            case EXITLISTENER:
                length = sprintf(mail, "^%i$", type);
                break;
            default:
                cout<<"invalid jobtype: "<<type<<endl;
        }
        DBG_PRINTF(("created message: %s\n", mail));
    }

    // add null terminator string length
    length++;

    return SAFE;
}


bool Peer::initThreads() {
    int status = SAFE;
    
    // only create all the threads once.
    if(!_rejoin){
      status = pthread_create(&_listenerContext, NULL, listener, this);

      for (int ii = 0; ii < maxWorkers; ii++) {
          pthread_t * newThreadContext = new pthread_t;
          status |= pthread_create(newThreadContext, NULL, worker, this);
          _workerContext.push_back(newThreadContext);
      }
    }

    status = pthread_create(&_shareContext, NULL, share, this);

    return (status == 0);
}

string Peer::convertInt(int number)
{
   stringstream ss;
   ss << number;
   return ss.str();
}

int Peer::insert(string filename){
    int status = SAFE;

    // verify file exists.
    ifstream ifile(filename.c_str());
    if (ifile) {
        // copy file to local directory
        string shortName = filename;
        // if the file has an path, then find the actual file name
        size_t backSlash = filename.rfind("/");
        if (backSlash != string::npos)
        {
            shortName = filename.substr(backSlash+1, filename.length() - backSlash);
        }

        string output = _localDirectory+"/"+shortName;
        ofstream ofile (output.c_str(), fstream::trunc|fstream::binary);
        ofile << ifile.rdbuf();   

        // calculate how many chunks this file has
        ifile.seekg (0, ios::end);
        long length = ifile.tellg();
        ifile.seekg (0, ios::beg);
        int chunks = length/chunkSize+1;

        // create file object on local status
        status = addFile(shortName, chunks, true);

        if(localInfo->State == peerInfo::connected && status == SAFE){
            // create BROADCAST_NEWFILE job
            NewFileArgs *jobArgs = new NewFileArgs();        
            jobArgs->fileName = shortName;
            //jobArgs->num_chunks = chunks; 

            Job *job = new Job();
            job->type = Job::BROADCAST_NEWFILE;
            job->args = (void *)jobArgs;

            // acquire the job queue mutex; enqueue; release mutex
            pushJob(job);

            // wake up share
            sem_post (&share_lock); 
        }

    } else {
        // file does not exist.
        status = errNoFileFound;
    }

    return status;
}

int Peer::writeChunk(string fileName, int chunkNum, char *content, int size){
    pthread_mutex_lock(&file_writing_mutex);
    string output = _localDirectory+"/"+fileName+".part"+convertInt(chunkNum);
    ofstream ofile (output.c_str(), fstream::trunc|ofstream::binary);
    ofile.write(content, size);
    ofile.close();
    pthread_mutex_unlock(&file_writing_mutex);
    return SAFE;
}

int Peer::combineFile(string fileName, int chunks){
    int status = SAFE;
    pthread_mutex_lock(&file_writing_mutex);
    DBG_PRINTF(("Combining all the file parts...\n"));
    
    DBG_PRINTF(("File %s with %i chunks\n", fileName.c_str(), chunks));

    string finaloutput = _localDirectory+"/"+fileName;
    // remove the file if exist
    ifstream file(finaloutput.c_str());
    if(file){
        if( remove( finaloutput.c_str() ) != 0 ){
            DBG_PRINTF(( "Error deleting file: %s\n", finaloutput.c_str() ));
        } else {
            DBG_PRINTF(( "File successfully deleted %s\n", finaloutput.c_str()  ));
        }
    }
    ofstream ofile (finaloutput.c_str(), fstream::trunc|ofstream::binary);

    for(int i = 1; i <= chunks; i++){
        string input = _localDirectory+"/"+fileName+".part"+convertInt(i);
        ifstream ifile(input.c_str());
        if(ifile){
            ofile<<ifile.rdbuf();
            // remove the partial file
            if( remove( input.c_str() ) != 0 ){
                DBG_PRINTF(( "Error deleting file: %s\n", input.c_str() ));
            } else {
                DBG_PRINTF(( "File successfully deleted %s\n", input.c_str()  ));
            }
        }else{
            status = errChunkNotFound;
            break;
        }
    }
    ofile.close();
    pthread_mutex_unlock(&file_writing_mutex);
    return status;
}

int Peer::setShare(bool state) {
    pthread_mutex_lock (&sharing_mutex);
    _sharing = state;
    pthread_mutex_unlock (&sharing_mutex);

    if (state == false) {
        // post semaphore to wake up share()
        sem_post(&share_lock);
    }

    return SAFE;
}


int Peer::query(Status& status){
    // fill status with local _status   
    pthread_mutex_lock (&status_mutex);
    status = _status;
    pthread_mutex_unlock (&status_mutex);
    return SAFE;
}



int Peer::initializePeersInfo(string peersFile){
    _numPeers = 0;
    _curIndex = 0;
    _numConnected = 0;

    pthread_mutex_init (&numConnected_mutex, NULL);
    pthread_mutex_init (&index_mutex, NULL);
    pthread_mutex_init (&peersArray_mutex, NULL);

    int num_peers = 0;
    ifstream myfile(peersFile.c_str());

    if (myfile.is_open()) {

        // parse peersFile
        string ip_addr;
        int port_num;
        while ( myfile >> ip_addr >> port_num )
        {
            DBG_PRINTF(("peer %i is %s:%i\n", num_peers, ip_addr.c_str(), port_num));
            _peers[num_peers] = new peerInfo();
            _peers[num_peers]->IP = ip_addr;
            _peers[num_peers]->Port = port_num;
            _peers[num_peers]->State = peerInfo::unknown;

            num_peers++;

            // error check to make sure num_peers does not exceed max
        }
        myfile.close();
    }
    else return errNoFileFound;

    _numPeers = num_peers;

    return SAFE;
}

int Peer::join(){
    int status;

    if ( localInfo->State == peerInfo::connected ) {
        return errAlreadyJoined;
    }
    
    // initialize some variables
    _sharing = false;
    
    // only init this if first time joining.
    if(!_rejoin){
      sem_init (&job_queue_count, 0, 0);
    } 
    
    // update localhost status
    localInfo->State = peerInfo::connected;

    // create JOIN job
    Job *job = new Job();
    job->type = Job::BROADCAST_JOIN;

    // acquire the job queue mutex; enqueue; release mutex
    pushJob(job);

    // create and initialize Peers object
    initializePeersInfo(_peerList);
    status = joinNetwork();
    if (status < 0 ) {
        return status;
    }
  

    initThreads();
    sem_post (&share_lock);

    DBG_PRINTF(("threads initialized\n"));

    DBG_PRINTF(("end of join()\n"));

    // set flag so that we know we have joined at least once
    _rejoin = true;

    return SAFE;
}

int Peer::leave(){
    // leave only if you are connected
    if(localInfo->State != peerInfo::connected) { return errNotConnected;}

    DBG_PRINTF(("cancel share\n"));
    pthread_cancel(_shareContext);

    // notify all peers of absence
    Job *leaveJob = new Job();
    leaveJob->type = Job::BROADCAST_LEAVE;
    leaveJob->args = NULL;
    pushJob(leaveJob);

    DBG_PRINTF(("leave called\n"));
    // cancel all threads with flag variable to ensure socket closing

    DBG_PRINTF(("waiting for listener to end...\n"));
    
    // don't end the listener
    // send an EXIT job to the listener and end it
    /* Job *job = new Job();
    job->type = Job::EXITLISTENER;
    job->args = NULL;
    pushJob(job);
    
    pthread_join(_listenerContext, NULL);
    DBG_PRINTF(("listener ended\n"));
    pthread_cancel(_listenerContext);

    // send EXIT job to all workers to end them
    for(int i = 0; i < maxWorkers; i++){
        Job *endworker = new Job();
        endworker->type = Job::EXITWORKER;
        endworker->args = NULL;
        pushJob(endworker); 
    }

    for (unsigned int ii = 0; ii < _workerContext.size(); ii++) {
        pthread_join(*_workerContext[ii], NULL);
    }

    // delete all thread pointers
    while (!_workerContext.empty()) {
        delete _workerContext.back();
        _workerContext.pop_back();
    }

*/
    // initialize the vector of files: wipe out the system status for all the files.
    // all threads should've been killed at this point.  no need to acquire mutex.

    pthread_mutex_lock (&status_mutex);
    _status.leaveNetwork();
    pthread_mutex_unlock (&status_mutex);

    // kill the objects in transit queue.
    
    pthread_mutex_lock (&transit_mutex);
    while(!transit.empty()){
        delete transit.back();
        transit.pop_back();
    }
    pthread_mutex_unlock (&transit_mutex);

    // change own status to disconnected
    localInfo->State = peerInfo::disconnected;

    return SAFE;
    
}

int Peer::changeLocalDirectory(string newDirectory){
    // verify directory exists

    _localDirectory = newDirectory;
    return SAFE;
}


int Peer::addFile(string fileName, int chunks, bool complete){
    int status = errUnknownWarning;
    pthread_mutex_lock (&status_mutex);
    status = _status.addFile(fileName, chunks, complete);
    pthread_mutex_unlock (&status_mutex);

    return status;
}

int Peer::addChunk(string fileName, int chunk){
    int status = errUnknownWarning;
    bool complete = false;
    int totalchunks = 0;
    pthread_mutex_lock (&status_mutex);
    DBG_PRINTF(("acquired mutex for status: FILE %s CHUNK %i\n", fileName.c_str(), chunk));   
    status = _status.receivedFileChunk(fileName, chunk);
    complete = _status.isFileComplete(fileName, totalchunks);
    DBG_PRINTF(("File %s  completion status: %i\n", fileName.c_str(), complete));
    pthread_mutex_unlock (&status_mutex);
    if(complete){
        combineFile(fileName, totalchunks);
    }
    return status;
}

int Peer::addSystemChunk(string fileName, int chunk){
    int status = errUnknownWarning;
    pthread_mutex_lock (&status_mutex);
    status = _status.receivedSystemChunk(fileName, chunk);
    pthread_mutex_unlock (&status_mutex);
    return status;
}

int Peer::removeSystemChunk(string fileName, int chunk){
    int status = errUnknownWarning;
    pthread_mutex_lock (&status_mutex);
    status = _status.removeSystemChunk(fileName, chunk);
    pthread_mutex_unlock (&status_mutex);
    return status;
}

int Peer::pushJob(Job * job){
    pthread_mutex_lock (&job_queue_mutex);
    DBG_PRINTF(("pushing job into queue: %i\n", job->type));
    jobs.push(job);
    sem_post (&job_queue_count);
    pthread_mutex_unlock (&job_queue_mutex);
    return SAFE;
}

int Peer::displayNumPeers(){
    if(_peers != NULL){
        int total = numPeers();
        int connected = numConnected();
        printf("Total Number of Peers: %i\n", total);
        printf("Total Number of Peers Connected on Network: %i\n", connected);
    }
    return SAFE;
}

// initialize does not need mutex
int Peer::joinNetwork(){
    // "ping" all peers

    int             orig_sock;           // Original socket in client
    struct sockaddr_in
        serv_adr;            // Internet addr of server
    struct hostent  *host;               // The host (server) info

    // loop through all online peers and tell them I have arrived!
    // create thread per peer
    DBG_PRINTF(("number of peers in network is: %i\n", _numPeers));
    for (int ii = 0; ii < _numPeers; ii++) {
        host = gethostbyname( _peers[ii]->IP.c_str() );       // Obtain host (server) info
        if (host == (struct hostent *) NULL ) {
            perror("gethostbyname ");
        } else {
            memset(&serv_adr, 0, sizeof( serv_adr));       // Clear structure
            serv_adr.sin_family = AF_INET;                 // Set address type
            memcpy(&serv_adr.sin_addr, host->h_addr, host->h_length);
            serv_adr.sin_port   = htons( _peers[ii]->Port );

            if ((orig_sock = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
                perror("generate error");
                return errSocketNotCreated;
            }  

            DBG_PRINTF(("attempting to connect to %s:%i\n", _peers[ii]->IP.c_str(), _peers[ii]->Port));

            // try to connect to another peer (this is blocking???)
            if (connect( orig_sock,(struct sockaddr *)&serv_adr,
                sizeof(serv_adr)) < 0) {
                _peers[ii]->State = peerInfo::disconnected;
                DBG_PRINTF(("peer %s:%i is offline!\n", _peers[ii]->IP.c_str(), _peers[ii]->Port));
            } else {
                _peers[ii]->State = peerInfo::connected;

                // increment _num_connected
                _numConnected++;

                DBG_PRINTF(("peer %s:%i is online!\n", _peers[ii]->IP.c_str(), _peers[ii]->Port));

                // create status query message
                char buf[bufSize];
                char msg[bufSize];
                char *field;
                char *brkt;
                int msgLen = 0;
                int responseLen = 0;
                int num_files, msgType;

                msgLen = sprintf(msg, "^%i$", Peer::QUERYSTATUS);
                DBG_PRINTF(("writing to server: %s\n", msg));
                write(orig_sock, msg, msgLen+1);

                // get response from server
                if (( responseLen = read(orig_sock, buf, bufSize)) > 0 ) {

                    if (responseLen < 3 || buf[0] != '^' || buf[responseLen-2] != '$') {
                        close(orig_sock);
                        return errInvalidMessage; // return errInvalidMessage error
                    }
                    
                    field = strtok_r(buf, "^%$", &brkt); // get message type
                    msgType = atoi(field);
                    DBG_PRINTF(("msg type: %i\n", msgType));

                    // assert msg type is REPLYSTATUS
                    if(msgType != Peer::REPLYSTATUS) {
                        close(orig_sock);
                        return errInvalidMessage; // return errInvalidMessage error
                    }


                    string ip = "";
                    int port_num = 0;
                    int storage_size = 0;

                    field = strtok_r(NULL, "^%$", &brkt); // get sender IP
                    ip += field;
                    DBG_PRINTF(("sender IP: %s\n", ip));

                    field = strtok_r(NULL, "^%$", &brkt); // get sender Port
                    port_num = atoi(field);
                    DBG_PRINTF(("sender Port: %i\n", port_num));

                    field = strtok_r(NULL, "^%$", &brkt); // get # hosted files
                    num_files = atoi(field);
                    DBG_PRINTF(("# hosted files: %i\n", num_files));

                    field = strtok_r(NULL, "^%$", &brkt); // get size of storage
                    storage_size = atoi(field);
                    DBG_PRINTF(("size of storage: %i\n", storage_size));

                    _peers[ii]->totalHostedFiles = num_files;
                    _peers[ii]->capacity = storage_size;

                    // add machine to hosts machine list if enough storage capacity
                    if (storage_size >= storageThreshold) {
                        pthread_mutex_lock(&host_machine_list_mutex);
                        hostMachineList.push_back(_peers[ii]);
                        pthread_mutex_unlock(&host_machine_list_mutex);
                    }

                }
            }
            close(orig_sock);
        }
    }

    return SAFE;
}   

// returns client_sock that wants the file
int Peer::queryPeers(char * message, int messageLen, void *This) {
    // uses socket connection to send to all peers in network
    DBG_PRINTF(("Peers::queryPeers: %s\n", message));

    int             orig_sock;           // Original socket in client
    struct sockaddr_in
        serv_adr;            // Internet addr of server
    struct hostent  *host;               // The host (server) info 

    bool setShareCalled = false;

    // loop through all online peers and tell them I have arrived!
    // create thread per peer
    for (int ii = 0; ii < _numPeers; ii++) {

        pthread_mutex_lock (&peersArray_mutex);
        pthread_mutex_lock (&index_mutex);

        if ( _peers[_curIndex]->State == peerInfo::connected ) {

            string IP = _peers[_curIndex]->IP;
            int port = _peers[_curIndex]->Port;

            pthread_mutex_unlock (&peersArray_mutex);
            pthread_mutex_unlock (&index_mutex);

            host = gethostbyname( IP.c_str() );       // Obtain host (server) info
            if (host == (struct hostent *) NULL ) {
                DBG_PRINTF(("host name not found for %s:%i\n", IP.c_str(), port));
                perror("gethostbyname ");
            } else {
                if ((orig_sock = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
                    perror("generate error");
                    return 3;
                }

                memset(&serv_adr, 0, sizeof( serv_adr));       // Clear structure
                serv_adr.sin_family = AF_INET;                 // Set address type
                memcpy(&serv_adr.sin_addr, host->h_addr, host->h_length);
                serv_adr.sin_port   = htons( port );

                // try to connect to another peer (this is blocking???)
                if (connect( orig_sock,(struct sockaddr *)&serv_adr,
                    sizeof(serv_adr)) < 0) {
                    DBG_PRINTF(("cannot connect to peer %s:%i\n", IP.c_str(), port));
                    perror("connect error");
                } else {
                    // write message
                    write(orig_sock, message, messageLen);

                    // read reply
                    char buf[bufSize];
                    int responseLen = 0;
                    char *field;
                    char *fileName, *brkt;
                    int fileChunkNum;
                    int msgType;

                    if (( responseLen = read(orig_sock, buf, bufSize)) > 0 ) {
                        // parse reply
                        if (responseLen < 3 || buf[0] != '^' || buf[responseLen-2] != '$') {
                            close(orig_sock);
                            return errInvalidMessage; // return errInvalidMessage error
                        }

                        DBG_PRINTF(("msg: %s\n", buf));

                        field = strtok_r(buf, "^%$", &brkt); // get message type
                        msgType = atoi(field);
                        DBG_PRINTF(("msg type: %i\n", msgType));

                        if (msgType == Peer::WANTCHUNK) {
                            DBG_PRINTF(("received WANTCHUNK\n"));

                            // increment the index so next time we don't try to share the same files to the same person first
                            pthread_mutex_lock (&index_mutex);
                            _curIndex = (_curIndex + 1) % _numPeers;
                            pthread_mutex_unlock (&index_mutex);

                            // set Peer->_sharing to false, and wake up Peer::share()
                            ((Peer *)This)->setShare(false);

                            setShareCalled = true;        

                            fileName = strtok_r(NULL, "^%$", &brkt); // get file name
                            DBG_PRINTF(("file name: %s\n", fileName));

                            field = strtok_r(NULL, "^%$", &brkt); // get chunks number
                            fileChunkNum = atoi(field);
                            DBG_PRINTF(("chunk #: %i\n", fileChunkNum));

                            char* chunk = new char[chunkSize];
                            int size = 0;
                            int status = ((Peer *)This)->getChunk(fileName, fileChunkNum, chunk, size);
                            if (status == SAFE){
                                write(orig_sock, chunk, size);
                                DBG_PRINTF(("finished transferring chunk\n"));
                            } else {
                                DBG_PRINTF(("failed to get file %s chunk %i\n", fileName, fileChunkNum));
                                // [*] retry???
                            }

                            close(orig_sock);
                            delete chunk;
                            break;
                        } 
                    }
                    close(orig_sock);
                }
            } 
        } else {
                pthread_mutex_unlock (&peersArray_mutex);
                pthread_mutex_unlock (&index_mutex);
        }

        pthread_mutex_lock (&index_mutex);
        _curIndex = (_curIndex + 1) % _numPeers;
        pthread_mutex_unlock (&index_mutex);
    }

    if (!setShareCalled) {
        // set Peer->_sharing to false, and wake up Peer::share()
        ((Peer *)This)->setShare(false);   
    }

    return SAFE;
}



// right now the message does do anything...pass in appropriate flags to construct message
// maybe a class for message construction
int Peer::broadcast(char *message, int messageLen) {
    // uses socket connection to send to all peers in network
    DBG_PRINTF(("Peer::broadcast: %s\n", message));

    int             orig_sock;           // Original socket in client
    struct sockaddr_in
        serv_adr;            // Internet addr of server
    struct hostent  *host;               // The host (server) info

    // loop through all online peers and tell them I have arrived!
    // create thread per peer
    for (int ii = 0; ii < _numPeers; ii++) {

        pthread_mutex_lock (&peersArray_mutex);

        if ( _peers[ii]->State == peerInfo::connected ) {

            string IP = _peers[ii]->IP;
            int port = _peers[ii]->Port;

            pthread_mutex_unlock (&peersArray_mutex);

            host = gethostbyname( IP.c_str() );       // Obtain host (server) info
            if (host == (struct hostent *) NULL ) {
                DBG_PRINTF(("host name not found for %s:%i\n", IP.c_str(), port));
                perror("gethostbyname ");
            } else {
                memset(&serv_adr, 0, sizeof( serv_adr));       // Clear structure
                serv_adr.sin_family = AF_INET;                 // Set address type
                memcpy(&serv_adr.sin_addr, host->h_addr, host->h_length);
                serv_adr.sin_port   = htons( port );

                if ((orig_sock = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
                    perror("generate error");
                    return 3;
                } 

                // try to connect to another peer (this is blocking???)
                if (connect( orig_sock,(struct sockaddr *)&serv_adr,
                    sizeof(serv_adr)) < 0) {
                    DBG_PRINTF(("cannot connect to peer %s:%i\n", IP.c_str(), port));
                    perror("connect error");
                } else {
                    // write message
                    write(orig_sock, message, messageLen);

                    // does not require a reply from other peers
                    close(orig_sock);
                }
            }
        } else {
            pthread_mutex_unlock (&peersArray_mutex);
        }
    }

    DBG_PRINTF(("BROADCAST DONE\n"));

    return SAFE;
}

peerInfo *Peer::getPeer(int index){
    if (index < _numPeers) {
        return _peers[index];
    }

    return NULL;
}

int Peer::numPeers(){
    return _numPeers;
}

int Peer::numConnected(){
    int num_connected;

    pthread_mutex_lock (&numConnected_mutex);
    num_connected = _numConnected;
    pthread_mutex_unlock (&numConnected_mutex);

    return num_connected;
}

int Peer::peerLeave(string ip, int port) {
    pthread_mutex_lock (&peersArray_mutex);

    for (int ii = 0; ii < _numPeers; ii++) {
        if (_peers[ii]->IP == ip && _peers[ii]->Port == port) {
            if (_peers[ii]->State == peerInfo::connected) {
                _peers[ii]->State = peerInfo::disconnected;

                // remove from hostMachineList
                list<peerInfo *>::iterator it;
                for (it = hostMachineList.begin(); it != hostMachineList.end(); ++it) {
                    if ( (*it)->IP == ip && (*it)->Port == port ) {
                        hostMachineList.erase(it);
                        break;
                    }
                }
            }
            break;
        }
    } 
    pthread_mutex_unlock (&peersArray_mutex);

    pthread_mutex_lock (&numConnected_mutex);
    _numConnected--;
    pthread_mutex_unlock (&numConnected_mutex);


    return SAFE;
}

int Peer::peerJoin(string ip, int port, int capacity) {
    pthread_mutex_lock (&peersArray_mutex);
    for (int ii = 0; ii < _numPeers; ii++) {
        if (_peers[ii]->IP == ip && _peers[ii]->Port == port) {
            if (_peers[ii]->State != peerInfo::connected) {
                _peers[ii]->State = peerInfo::connected;
                _peers[ii]->capacity = capacity;

                // add to hostMachineList if sufficient storage capacity
                if (_peers[ii]->capacity >= storageThreshold) {
                    pthread_mutex_lock(&host_machine_list_mutex);
                    hostMachineList.push_front(_peers[ii]);
                    pthread_mutex_unlock(&host_machine_list_mutex);
                }
            }
            break;
        }
    } 
    pthread_mutex_unlock (&peersArray_mutex);

    pthread_mutex_lock (&numConnected_mutex);
    _numConnected++;
    pthread_mutex_unlock (&numConnected_mutex);

    

    return SAFE;
}

int Peer::peerNewFile(string ip, int port) {
    pthread_mutex_lock (&peersArray_mutex);
    for (int ii = 0; ii < _numPeers; ii++) {
        if (_peers[ii]->IP == ip && _peers[ii]->Port == port) {
            _peers[ii]->totalHostedFiles++;
            break;
        }
    } 
    pthread_mutex_unlock (&peersArray_mutex);

    return SAFE;
}

int Peer::updatePeerStatus(string ip, int port, int totalFiles, int capacity){
    pthread_mutex_lock (&peersArray_mutex);
    for (int ii = 0; ii < _numPeers; ii++) {
        if (_peers[ii]->IP == ip && _peers[ii]->Port == port) {
            _peers[ii]->totalHostedFiles = totalFiles;
            _peers[ii]->capacity = capacity;
            break;
        }
    } 
    pthread_mutex_unlock (&peersArray_mutex);

    return SAFE;
}


